<?php

/**
 * Created by PhpStorm.
 * @description:
 * @time:
 * @Author: yfl
 * @QQ 554665488
 * Date: 2018-5-29
 * Time: 18:20
 */
$GLOBALS['var'] = 'swoole学习';

class workerServer
{
    protected $swoole_server;
    protected $test;
    protected $table;
    protected $task_worker_num;

    public function __construct($config)
    {
        $this->swoole_server = new Swoole\Server('0.0.0.0', 9501, SWOOLE_PROCESS, SWOOLE_SOCK_TCP);
        $this->test = 'aaaaa' . PHP_EOL;
        $this->swoole_server->set($config);
        $this->swoole_server->on('start', [$this, 'onStart']);
        $this->swoole_server->on('WorkerStart', [$this, 'onWorkerStart']);
        $this->swoole_server->on('ManagerStart', [$this, 'onManagerStart']);
        $this->swoole_server->on('connect', [$this, 'onConnect']);
        $this->swoole_server->on('receive', [$this, 'onReceive']);
        $this->swoole_server->on('task', [$this, 'onTask']);
        $this->swoole_server->on('finish', [$this, 'onFinish']);
        $this->swoole_server->on('close', [$this, 'onClose']);
//        $GLOBALS['var']='glob变量修改';
        $this->task_worker_num = $this->swoole_server->setting['task_worker_num'];
        $this->table = new Swoole\Table(1024);
        $this->table->column('task_worker_id', Swoole\Table::TYPE_INT, 1);//task_worker id号
        $this->table->column('status', Swoole\Table::TYPE_INT, 1);
        $this->table->create();
        $this->swoole_server->start();
    }

    public function onStart(Swoole\Server $server)
    {
//        $this->table->set(1,['task_work_id'=>0,'status'=>1]);
//        echo 'onStart==' . $this->test;
        for ($i = 0; $i < $this->task_worker_num; $i++) {
            $this->table->set($i, ['task_worker_id' => $i, 'status' => 1]);//设置每一个task_worker进程的ID的工作状态 1代表空闲状态 2代表工作状态
        }
    }

    public function onManagerStart(Swoole\Server $server)
    {
//        var_dump($this->table->get(1));
//        $this->table->set(1,['task_work_id'=>2,'status'=>2]);
//        echo 'onMangerStart==' . $this->test;
    }

    public function onWorkerStart(Swoole\Server $server, $worker_id)
    {

//        echo 'onWorkerStart==' . $this->test;
    }

    public function onConnect(Swoole\Server $server, $fd, $from_id)
    {
//        var_dump($this->table->get(1));
//        echo 'onConnect==' . $this->test;
    }

    public function onReceive(Swoole\Server $server, $fd, $from_id, $data)
    {
//        echo $GLOBALS['var'];
//        echo 'onReceive==' . $this->test;
        $data = [];
        for ($i = 0; $i < 20; $i++) {
            $data[] = '大量的数据需要很长的处理时间';
        }
        //保存空闲的task_worker_id 的状态
        $freeTask = [];
        foreach ($this->table as $item) {
            var_dump($item);
            if ($item['status'] == 1) {
                $freeTask[]['task_worker_id'] = $item['task_worker_id'];
            }
        }
//        echo '空闲的状态task_worker_id';
//        var_dump($freeTask);
        $freeTaskCount = count($freeTask); // 空闲的数量
        //没有全部空闲处理
        echo '一共的task_worker_num' . $this->task_worker_num;
        echo '空闲的状态数量'.$freeTaskCount;
        if ($freeTaskCount > 0 && $freeTaskCount != $this->task_worker_num) {
            //模拟数据切割分批数据投递到不同的task_work 进程中
            $count = count($data);
            $data = array_chunk($data, ceil($count / $freeTaskCount)); //数据除以task空闲状态id
//            var_dump($freeTaskCount);
            foreach ($data as $k => $v) {
//                $v['task_worker_id'] = $freeTaskCount[$k]['task_worker_id']; //保存当前任务,来自于哪个task_worker_id
                $v['task_worker_id'] = $freeTask[$k]['task_worker_id']; //保存当前任务,来自于哪个task_worker_id
                $server->task($v, $freeTask[$k]['task_worker_id']);
                echo '我是空闲状态的投递=========================================' . $freeTask[$k]['task_worker_id'];
            }
        } else {
            //全部空闲处理
            //模拟数据切割分批数据投递到不同的task_work 进程中
            //平均分配给相应的task进程
            //数据平均分割成task_worker_num份,交给task_worker_num个工作进程
            $count = count($data);
            $data = array_chunk($data, ceil($count / $this->task_worker_num));
//            var_dump($data);
            //        echo '工作进程的数量'.$this->task_worker_num;
            //        echo '切割的数据数量'.ceil($count / $this->task_worker_num);
            foreach ($data as $k => $v) {
                $v['task_worker_id'] = $k; //保存当前任务,来自于哪个task_worker_id
                $server->task($v, $k);
            }
        }
        echo '我是先执行task_worker任务去异步执行';
//        sleep(5);
    }

    public function onTask(Swoole\Server $server, $task_id, $src_worker_id, $data)
    {
        echo '当前等待的task任务数量' . $server->stats()['tasking_num'] . PHP_EOL;
        echo '消息任务编号为' . $task_id . '来自worker进程的id=' . $src_worker_id . PHP_EOL;
        $this->table->incr($data['task_worker_id'], 'status', 1);
        if($server->stats()['tasking_num']==0) echo  '执行任务完成';
        $rand = mt_rand(1, 10);
        sleep($rand);
        return ['finish_task_worker' => $data['task_worker_id']]; // 把完成任务的task_worker_id投递到finish worker进程中用来修改状态
    }

    public function onFinish(Swoole\Server $server, $task_id, $data)
    {
//        var_dump($data);
//        var_dump($this->table);

        $this->table->decr($data['finish_task_worker'], 'status', 1);
    }

    public function onClose(Swoole\Server $server)
    {

    }
}

$config = [
    'worker_num' => 2,
    'reactor_num' => 2,
    'task_worker_num' => 4,
    'task_max_request' => 5000,
    'max_request' => 1024,
    // 心跳检测
    'heartbeat_check_interval' => 60,
    'heartbeat_idle_time' => 130,
    //包头包体检测
    'open_length_check' => true,
    'package_length_type' => 'N',
    'package_body_offset' => 4,
    'package_length_offset' => 0,
    'package_max_length' => 1024 * 1024 * 3 //设置包的最长字节
];
new workerServer($config);